package com.yukefms.message;

import com.yukefms.engine.EngineConfig;
import com.yukefms.engine.SimpleFMSEngine;
import org.apache.kafka.common.utils.Java;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Component;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Description
 * @Author Ping
 * @Date 2021/6/25 15:07
 **/
@Component
public class MessageContainer<T> {

    public static Logger logger = LogManager.getLogger(MessageContainer.class);

    private Queue<T> innerCache ;
    /**
     *
     */
    private MessageConsumer messageConsumer ;

    private int state =  INACTIVED ;
    /**
     * containter states
     */
    public static final int INACTIVED = 0 ;
    public static final int STARTED = 1 ;
    public static final int STOPED = 2 ;


    public MessageContainer() {
        super() ;
        this.innerCache = new ConcurrentLinkedQueue<>() ;
        //this.innerCache = new LinkedBlockingQueue<>() ;
        this.state = INACTIVED ;
    }
    public boolean empty() {
        return this.innerCache.isEmpty() ;
    }
    public void add(T t) {
        this.innerCache.add(t) ;
    }
    public int size() {
        return this.innerCache.size() ;
    }
    public boolean overLimit() {
        return this.innerCache.size() > EngineConfig.CONTENT_LIMIT ;
    }
    /**
     * get and remove the head of container.
     * @return
     */
    public T pop() {
        // NOTE THAT, here needs to consider the exceptonal situation.
        if ( this.ended()) {
            System.out.println("[MessageContainter] container is stop work. "  ) ;
            return null ;
        }
//System.out.println("[MessageContainter] the size in the container is : "  + this.innerCache.size()) ;
        return this.innerCache.poll() ;
    }


    /**
     * start the message consumer reading work
     * @return
     */
    public int startWork() {
        try {
            if (this.messageConsumer == null) {
                this.messageConsumer = new MessageConsumer((MessageContainer<String>)this) ;
                logger.error("the message consumer in the Container was created successfully.") ;
            }
            new Thread(this.messageConsumer).start() ;
            this.state = STARTED ;
            logger.error("the message container was started up successfully.") ;
        } catch (RuntimeException e) {
            logger.error("the message container was started up failed.") ;
            return -1 ;
        }
        return 0 ;
    }

    /**
     * stop message consumer work
     */
    public void stopWork() {
        this.messageConsumer.stop() ;
        this.state = STOPED ;
        logger.info("the message container was stoped successfully.") ;
    }
    public boolean ended() {
        return this.state == STOPED || this.messageConsumer.isEnded() ;
    }
    public boolean isStarted() {
        return this.state == STARTED ;
    }


}
